package com.hm.manage.utils;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
import org.springframework.stereotype.Component;

import java.math.BigInteger;
import java.util.*;

@Slf4j
@Component
public class KafkaOffsetLagFetcher {

    /***
     * 获取消费组的堆积情况
     * @param groupId
     * @param topic
     * @return
     */
    public Map<Integer,Long> getLagMap(String bootstrapServers,String groupId, String topic) {
        HashMap<Integer, Long> lagmap = new HashMap<>();
//        String bootstrapServers = "50.58.191.82:9092";
//        groupId = "connector-chain-bayonet_vehiclepass";
//        topic = "BAYONET_VEHICLEPASS";

        Properties props = new Properties();
        props.put("bootstrap.servers", bootstrapServers);
        props.put("group.id", groupId);
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("enable.auto.commit", "false");

        try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
            // 获取主题的所有分区
            List<PartitionInfo> partitions = consumer.partitionsFor(topic);
            List<TopicPartition> topicPartitions = new ArrayList<>();
            for (PartitionInfo partition : partitions) {
                topicPartitions.add(new TopicPartition(topic, partition.partition()));
            }

            // 获取已提交的offset
            Map<TopicPartition, OffsetAndMetadata> committedOffsets = consumer.committed(new HashSet<>(topicPartitions));

            // 获取最新的offset
            Map<TopicPartition, Long> endOffsets = consumer.endOffsets(topicPartitions);

            // 计算并打印lag
            for (TopicPartition tp : topicPartitions) {
                long endOffset = endOffsets.get(tp);
                OffsetAndMetadata committed = committedOffsets.get(tp);
                long committedOffset = (committed != null) ? committed.offset() : 0;
                long lag = endOffset - committedOffset;
                lagmap.put(tp.partition(),lag);
                System.out.printf("分区 %d: 当前offset=%d, 最新offset=%d, lag=%d%n",
                        tp.partition(), committedOffset, endOffset, lag);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return lagmap;
    }

    public String checkLagMap(Map<Integer, Long> lagmap){
        if (lagmap.isEmpty()){
            log.error("----------lagmap为空-----------");
            return null;
        }
        ArrayList<Long> values = new ArrayList<>(lagmap.values());

        // 获取最大和最小
        Long max = Collections.max(values);
        Long min = Collections.min(values);

        // 获取平均值
        BigInteger sum = BigInteger.ZERO;
        for (Long value : values){
            sum = sum.add(BigInteger.valueOf(value));
        }
        double average = sum.doubleValue() / values.size();
        log.info("-------堆积最大：{}|最小：{}|平均：{}------------------",max,min,average);
        return "堆积最大:"+max+",最小:"+min+",平均:"+average;
    }

    /***
     * 获取最大堆积
     * @param lagmap
     * @return
     */
    public Long getMaxLag(Map<Integer, Long> lagmap){
        if (lagmap.isEmpty()){
            log.error("----------lagmap为空-----------");
            return 0L;
        }
        ArrayList<Long> values = new ArrayList<>(lagmap.values());
        Long max = Collections.max(values);
        return max;
    }

    /****
     * 获取堆积平均值
     * @param lagmap
     * @return
     */
    public Double getAverageLag(Map<Integer, Long> lagmap){
        if (lagmap.isEmpty()){
            log.error("----------lagmap为空-----------");
            return 0.0;
        }
        ArrayList<Long> values = new ArrayList<>(lagmap.values());
        // 获取平均值
        BigInteger sum = BigInteger.ZERO;
        for (Long value : values){
            sum = sum.add(BigInteger.valueOf(value));
        }
        double average = sum.doubleValue() / values.size();
        return average;
    }

}
